<?php declare(strict_types=1);


namespace App\Listener;
use App\Rpc\Lib\MessageInterface;
use Six\Rpc\Client\Annotation\Mapping\Reference;
use Six\Rpc\Client\ServiceContext;
use Swoft\Bean\Annotation\Mapping\Inject;
use Swoft\Event\Annotation\Mapping\Listener;
use Swoft\Event\EventHandlerInterface;
use Swoft\Event\EventInterface;
use Swoft\Redis\Pool;
use Swoft\Server\ServerEvent;

/**
 * Class RegisterServer
 * @package App\Listener
 * @Listener(ServerEvent::BEFORE_START)
 */
class RegisterServer implements EventHandlerInterface
{

    /**
     * @Inject("redis.pool")
     * @var Pool
     */
    private $connectionRedis;

    /**
     * @Inject("rabbit.pool")
     * @var \Six\Rabbit\Pool
     */
    private $rabbit;


    /**
     * @Reference(pool="message.pool",fallback="MessageFallback")
     *
     * @var MessageInterface
     */
    private $messageService;

    /**
     * 独立的事件监听器接口，在服务启动前注册服务
     * @param EventInterface $event
     * @throws \ReflectionException
     * @throws \Swoft\Bean\Exception\ContainerException
     */
    public function handle(EventInterface $event) : void
    {
        //注册consul服务
        var_dump('服务启动前consul服务中注册服务');
        //获取consul配置信息
        $config = bean('config')->get('provider.consul');
        //注册consul配置
        bean('consulProvider')->registerServer($config);

        #------------------------监听消息-------------------------


        $callBack=function (){
            sgo(function (){
                //监听消息捕获（消费者）
                $context = ServiceContext::new();
                \Swoft\Context\Context::set($context);

                $exchangeName = "tradeExchange";#通道名称
                $routeKey = "/trade";
                $queueName = 'trade';
                $connection=$this->rabbit->connect();
                $connectionRabbit=$connection->connection;
                $channel = $connectionRabbit->channel();

                #声明队列
                $channel->queue_declare($queueName, false, true, false, false);
                #声明交换机
//                $channel->exchange_declare($exchangeName, \PhpAmqpLib\Exchange\AMQPExchangeType::DIRECT, false, true, false);
                $channel->exchange_declare($exchangeName, \PhpAmqpLib\Exchange\AMQPExchangeType::DIRECT, true, false, null);

                //队列绑定交换机跟路由
                $channel->queue_bind($queueName, $exchangeName, $routeKey);



                $channel->basic_consume($queueName,'',false,false,false,false,function($message){

                    $data = json_decode($message->body ,true );
                    #1.记录消息任务是否完成（消费幂等:同一个任务执行10次跟执行一次的效果是一样）
                    $statusJob = $this->connectionRedis->get("integrating_message_job", (string)$data['msg_id']);

                    if($statusJob == 1){
                        $this->messageService->ackMsg($data['msg_id']); //这个任务已经消费成功了
                    }elseif ($statusJob == 0){
                        var_dump('任务正在执行当中');
                        return;
                    }else{
                        //执行任务当中,并且设置释放的时间
                        $this->connectionRedis->setex("integrating_message_job:".$data['msg_id'],10,1);
                        //sleep(5);//任务正在执行当中
                        //2.操作mysql更新积分(业务逻辑执行完毕)
                        $this->connectionRedis->set("integrating_message_job:".$data['msg_id'] , 2);//执行任务完毕
                        $this->messageService->ackMsg($data['msg_id']); //这个任务已经消费成功了
                        //回应ack
                        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
                    }
                });

                // Loop as long as the channel has callbacks registered
                while ($channel->is_consuming()) {
                    $channel->wait(); //阻塞消费
                }

                /**
                 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
                 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
                 */
                function shutdown($channel, $connection)
                {
                    $channel->close();
                    $connection->close();
                }
                register_shutdown_function('shutdown', $channel, $this->connectionRedis);
            });



        };

        //        //防止进程意外崩溃，回收子进程
        \Swoole\Process::signal(SIGCHLD, function ($sig) use ($callBack) {
            while ($ret = \Swoole\Process::wait(false)) {
                $p = new  \Swoole\Process($callBack);
                $p->start();
            }
        });

        $p = new  \Swoole\Process($callBack);
        $p->start();
    }
}